#include "asycipc/named_pipe.hpp"

#ifdef __APPLE__

#include <fcntl.h>
#include <signal.h>
#include <sys/event.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cctype>
#include <cmath>
#include <cstdlib>
#include <cstring>
#include <filesystem>
#include <iostream>
#include <limits>
#include <map>
#include <mutex>
#include <queue>
#include <sstream>
#include <stdexcept>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>

#include "asycipc/api.hpp"
#include "asycipc/frame.hpp"
#include "asycipc/linux_diagnostics.hpp"
#include "asycipc/server_runtime.hpp"
#include "json11.hpp"

namespace asycipc {

namespace {

using json11::Json;
namespace fs = std::filesystem;

constexpr mode_t kDirectoryMode = 0700;
constexpr mode_t kFifoMode = 0600;
constexpr std::chrono::seconds kHealthScanInterval(5);
constexpr std::chrono::seconds kHealthAlertInterval(30);

bool MacDebugEnabled() {
  static const bool enabled = []() {
    if (const char* flag = std::getenv("ASYIPC_MAC_DEBUG")) {
      return *flag != '\0' && std::string(flag) != "0";
    }
    return false;
  }();
  return enabled;
}

void MacDebugLog(const std::string& message) {
  if (!MacDebugEnabled()) {
    return;
  }
  std::cerr << "[asycipc][mac] " << message << std::endl;
}

class PosixError : public NamedPipeError {
 public:
  PosixError(const std::string& where, int err)
      : NamedPipeError(where + " failed: " + std::string(std::strerror(err))) {}
};

void ThrowPosixError(const std::string& where) {
  throw PosixError(where, errno);
}

void EnsureNotSymlink(const fs::path& path) {
  struct stat st {};
  if (lstat(path.c_str(), &st) != 0) {
    if (errno == ENOENT) {
      return;
    }
    ThrowPosixError("lstat(" + path.string() + ")");
  }
  if (S_ISLNK(st.st_mode)) {
    throw NamedPipeError("symlink detected at " + path.string());
  }
}

void EnsureDirectorySecure(const fs::path& path, mode_t mode) {
  EnsureNotSymlink(path);
  if (!fs::exists(path)) {
    std::error_code ec;
    fs::create_directories(path, ec);
    if (ec) {
      throw NamedPipeError("create_directories failed for " + path.string() + ": " +
                           ec.message());
    }
  }

  struct stat st {};
  if (lstat(path.c_str(), &st) != 0) {
    ThrowPosixError("lstat(" + path.string() + ")");
  }
  if (!S_ISDIR(st.st_mode)) {
    throw NamedPipeError(path.string() + " is not a directory");
  }
  if (st.st_uid != geteuid()) {
    throw NamedPipeError("directory " + path.string() + " not owned by current user");
  }
  const mode_t current_mode = st.st_mode & 0777;
  if (current_mode != mode) {
    std::cerr << "[asycipc] adjusting permissions on directory " << path
              << " from 0" << std::oct << current_mode << " to 0" << mode << std::dec
              << std::endl;
    if (chmod(path.c_str(), mode) != 0) {
      ThrowPosixError("chmod(" + path.string() + ")");
    }
  }
}

void EnsureFifoSecure(const fs::path& path, mode_t mode) {
  EnsureNotSymlink(path);
  struct stat st {};
  if (lstat(path.c_str(), &st) != 0) {
    if (errno == ENOENT) {
      if (mkfifo(path.c_str(), mode) != 0) {
        ThrowPosixError("mkfifo(" + path.string() + ")");
      }
      return;
    }
    ThrowPosixError("lstat(" + path.string() + ")");
  }
  if (!S_ISFIFO(st.st_mode)) {
    throw NamedPipeError(path.string() + " is not a FIFO");
  }
  if (st.st_uid != geteuid()) {
    throw NamedPipeError("fifo " + path.string() + " not owned by current user");
  }
  const mode_t current_mode = st.st_mode & 0777;
  if (current_mode != mode) {
    std::cerr << "[asycipc] adjusting permissions on fifo " << path
              << " from 0" << std::oct << current_mode << " to 0" << mode << std::dec
              << std::endl;
    if (chmod(path.c_str(), mode) != 0) {
      ThrowPosixError("chmod(" + path.string() + ")");
    }
  }
}

fs::path DetermineRuntimeRoot() {
  if (const char* override_dir = std::getenv("ASYIPC_RUNTIME_DIR")) {
    if (*override_dir != '\0') {
      return fs::path(override_dir);
    }
  }

  if (geteuid() == 0) {
    return fs::path("/var/run/asycipc");
  }

  if (const char* xdg_dir = std::getenv("XDG_RUNTIME_DIR")) {
    if (*xdg_dir != '\0') {
      return fs::path(xdg_dir) / "asycipc";
    }
  }

  const char* tmp = std::getenv("TMPDIR");
  if (tmp && *tmp) {
    return fs::path(tmp) / "asycipc";
  }

  fs::path user_run = fs::temp_directory_path() / "asycipc";
  return user_run;
}

fs::path ResolveBasePath(const std::string& endpoint) {
  if (endpoint.empty()) {
    throw NamedPipeError("named pipe endpoint must not be empty");
  }

  fs::path base;
  if (!endpoint.empty() && endpoint[0] == '/') {
    base = fs::path(endpoint);
  } else {
    base = DetermineRuntimeRoot() / endpoint;
  }

  EnsureDirectorySecure(base, kDirectoryMode);
  return base;
}

bool IsSessionIdValid(const std::string& id) {
  if (id.empty() || id.size() > 64) {
    return false;
  }
  for (char ch : id) {
    if ((ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') ||
        (ch >= '0' && ch <= '9') || ch == '-' || ch == '_') {
      continue;
    }
    return false;
  }
  return true;
}

std::string SanitizeSessionId(const std::string& source, const std::string& fallback) {
  std::string id;
  id.reserve(source.size());
  for (unsigned char ch : source) {
    if (std::isalnum(ch)) {
      id.push_back(static_cast<char>(std::tolower(ch)));
    } else if (ch == '-' || ch == '_') {
      if (!id.empty() && id.back() == '_') {
        continue;
      }
      id.push_back(ch);
    } else {
      if (!id.empty() && id.back() == '_') {
        continue;
      }
      id.push_back('_');
    }
  }
  if (id.empty()) {
    id = fallback;
  }
  if (id.size() > 48) {
    id = id.substr(id.size() - 48);
  }
  return id;
}

struct MacHandshakeRequest {
  std::string session_id;
  pid_t client_pid = -1;
};

MacHandshakeRequest ParseHandshake(const std::string& payload) {
  std::string err;
  const Json json = Json::parse(payload, err);
  if (!err.empty()) {
    throw NamedPipeError("invalid handshake message: " + err);
  }
  if (json.type() != Json::OBJECT) {
    throw NamedPipeError("handshake message must be object");
  }

  MacHandshakeRequest request;
  const auto& obj = json.object_items();
  auto it = obj.find("session");
  if (it == obj.end() || it->second.type() != Json::STRING) {
    throw NamedPipeError("handshake missing session id");
  }
  request.session_id = it->second.string_value();
  if (!IsSessionIdValid(request.session_id)) {
    throw NamedPipeError("invalid session identifier");
  }

  auto pid_it = obj.find("pid");
  if (pid_it != obj.end()) {
    if (pid_it->second.type() != Json::NUMBER) {
      throw NamedPipeError("handshake pid must be numeric");
    }
    const double pid_value = pid_it->second.number_value();
    if (!std::isfinite(pid_value) || pid_value < 0 ||
        pid_value > static_cast<double>(std::numeric_limits<pid_t>::max())) {
      throw NamedPipeError("handshake pid out of range");
    }
    request.client_pid = static_cast<pid_t>(pid_value);
  }

  return request;
}

class MacPipeHandle : public PipeHandle {
 public:
  MacPipeHandle(int read_fd, int write_fd, fs::path read_path, fs::path write_path)
      : read_fd_(read_fd), write_fd_(write_fd), read_path_(std::move(read_path)),
        write_path_(std::move(write_path)) {}

  ~MacPipeHandle() override { Close(); }

  void Write(const std::vector<uint8_t>& data) override {
    if (write_fd_ < 0) {
      throw NamedPipeError("pipe closed");
    }
    if (MacDebugEnabled()) {
      MacDebugLog("write begin path=" + write_path_.string() +
                  " bytes=" + std::to_string(data.size()));
    }
    const uint8_t* cursor = data.data();
    std::size_t remaining = data.size();
    while (remaining > 0) {
      ssize_t written = ::write(write_fd_, cursor, remaining);
      if (written < 0) {
        if (MacDebugEnabled()) {
          MacDebugLog("write error path=" + write_path_.string() +
                      " errno=" + std::to_string(errno) +
                      " msg=" + std::string(std::strerror(errno)));
        }
        if (errno == EINTR) {
          continue;
        }
        ThrowPosixError("write");
      }
      if (written == 0) {
        throw NamedPipeError("zero-byte write detected");
      }
      remaining -= static_cast<std::size_t>(written);
      cursor += written;
      if (MacDebugEnabled()) {
        MacDebugLog("write progress path=" + write_path_.string() +
                    " chunk=" + std::to_string(written) +
                    " remaining=" + std::to_string(remaining));
      }
    }
    if (MacDebugEnabled()) {
      MacDebugLog("write complete path=" + write_path_.string());
    }
  }

  std::vector<uint8_t> Read() override {
    if (read_fd_ < 0) {
      throw NamedPipeError("pipe closed");
    }
    while (true) {
      ssize_t received = ::read(read_fd_, read_buffer_.data(), read_buffer_.size());
      if (received < 0) {
        if (MacDebugEnabled()) {
          MacDebugLog("read error path=" + read_path_.string() +
                      " errno=" + std::to_string(errno) +
                      " msg=" + std::string(std::strerror(errno)));
        }
        if (errno == EINTR) {
          continue;
        }
        ThrowPosixError("read");
      }
      if (received == 0) {
        if (MacDebugEnabled()) {
          MacDebugLog("read eof path=" + read_path_.string());
        }
        return {};
      }
      read_scratch_.assign(read_buffer_.data(),
                           read_buffer_.data() + static_cast<std::size_t>(received));
      if (MacDebugEnabled()) {
        MacDebugLog("read complete path=" + read_path_.string() +
                    " bytes=" + std::to_string(received));
      }
      return read_scratch_;
    }
  }

  int read_fd() const noexcept { return read_fd_; }

  void Close() {
    if (read_fd_ >= 0) {
      ::close(read_fd_);
      read_fd_ = -1;
    }
    if (write_fd_ >= 0) {
      ::close(write_fd_);
      write_fd_ = -1;
    }
  }

 private:
  int read_fd_ = -1;
  int write_fd_ = -1;
  fs::path read_path_;
  fs::path write_path_;
  std::array<uint8_t, 4096> read_buffer_{};
  std::vector<uint8_t> read_scratch_;
};

class MacPipeListener : public PipeListener {
 public:
  explicit MacPipeListener(fs::path base_dir)
      : base_dir_(std::move(base_dir)), listener_path_(base_dir_ / "listener") {
    EnsureDirectorySecure(base_dir_, kDirectoryMode);
    EnsureFifoSecure(listener_path_, kFifoMode);
  }

  PipeHandlePtr Accept() override {
    while (true) {
      std::string handshake;
      if (ReadHandshake(handshake)) {
        return CompleteAccept(handshake);
      }
    }
  }

  const fs::path& base_dir() const { return base_dir_; }

  void SignalInterrupt() const {
    const char byte = 0;
    if (keepalive_fd_ >= 0) {
      ::write(keepalive_fd_, &byte, 1);
      return;
    }
    int fd = ::open(listener_path_.c_str(), O_WRONLY | O_NONBLOCK | O_CLOEXEC);
    if (fd < 0) {
      return;
    }
    ::write(fd, &byte, 1);
    ::close(fd);
  }

 private:
  bool ReadHandshake(std::string& line) {
    if (listener_fd_ < 0) {
      listener_fd_ = ::open(listener_path_.c_str(), O_RDONLY | O_CLOEXEC);
      if (listener_fd_ < 0) {
        ThrowPosixError("open(" + listener_path_.string() + ")");
      }
      keepalive_fd_ = ::open(listener_path_.c_str(), O_WRONLY | O_CLOEXEC | O_NONBLOCK);
      if (keepalive_fd_ < 0 && errno != ENXIO) {
        ThrowPosixError("open(keepalive)");
      }
    }

    while (true) {
      auto pos = handshake_buffer_.find('\n');
      if (pos != std::string::npos) {
        line.assign(handshake_buffer_.data(), pos);
        handshake_buffer_.erase(0, pos + 1);
        return true;
      }

      char chunk[256];
      ssize_t n = ::read(listener_fd_, chunk, sizeof(chunk));
      if (n < 0) {
        if (errno == EINTR) {
          continue;
        }
        ThrowPosixError("read(listener)");
      }
      if (n == 0) {
        ::close(listener_fd_);
        listener_fd_ = -1;
        if (keepalive_fd_ >= 0) {
          ::close(keepalive_fd_);
          keepalive_fd_ = -1;
        }
        handshake_buffer_.clear();
        return false;
      }
      handshake_buffer_.append(chunk, static_cast<std::size_t>(n));
    }
  }

  PipeHandlePtr CompleteAccept(const std::string& handshake) {
    MacHandshakeRequest request = ParseHandshake(handshake);

    const fs::path in_path = base_dir_ / ("session-" + request.session_id + "-in");
    const fs::path out_path = base_dir_ / ("session-" + request.session_id + "-out");
    if (MacDebugEnabled()) {
      MacDebugLog("accept session=" + request.session_id + " in_path=" + in_path.string() +
                  " out_path=" + out_path.string());
    }

    EnsureFifoSecure(in_path, kFifoMode);
    EnsureFifoSecure(out_path, kFifoMode);

    int read_fd = ::open(in_path.c_str(), O_RDONLY | O_CLOEXEC);
    if (read_fd < 0) {
      ThrowPosixError("open(" + in_path.string() + ")");
    }

    int write_fd = ::open(out_path.c_str(), O_WRONLY | O_CLOEXEC);
    if (write_fd < 0) {
      ::close(read_fd);
      ThrowPosixError("open(" + out_path.string() + ")");
    }
    if (MacDebugEnabled()) {
      MacDebugLog("accept opened session=" + request.session_id +
                  " read_fd=" + std::to_string(read_fd) +
                  " write_fd=" + std::to_string(write_fd));
    }

    return std::make_unique<MacPipeHandle>(read_fd, write_fd, in_path, out_path);
  }

  fs::path base_dir_;
  fs::path listener_path_;
  mutable int listener_fd_ = -1;
  mutable int keepalive_fd_ = -1;
  mutable std::string handshake_buffer_;
};

void WriteAll(int fd, const std::string& data) {
  const char* cursor = data.data();
  std::size_t remaining = data.size();
  while (remaining > 0) {
    ssize_t written = ::write(fd, cursor, remaining);
    if (written < 0) {
      if (errno == EINTR) {
        continue;
      }
      ThrowPosixError("write");
    }
    if (written == 0) {
      throw NamedPipeError("zero-byte write detected");
    }
    remaining -= static_cast<std::size_t>(written);
    cursor += written;
  }
}

void DrainPipe(int fd) {
  char buffer[64];
  while (true) {
    ssize_t n = ::read(fd, buffer, sizeof(buffer));
    if (n < 0) {
      if (errno == EINTR) {
        continue;
      }
      if (errno == EAGAIN || errno == EWOULDBLOCK) {
        return;
      }
      ThrowPosixError("read(event pipe)");
    }
    if (n == 0 || n < static_cast<ssize_t>(sizeof(buffer))) {
      return;
    }
  }
}

}  // namespace

std::unique_ptr<PipeListener> CreateNamedPipeListener(
    const NamedPipeListenerOptions& options) {
  fs::path base_dir = ResolveBasePath(options.endpoint);
  return std::make_unique<MacPipeListener>(std::move(base_dir));
}

PipeHandlePtr ConnectNamedPipeEndpoint(
    const NamedPipeConnectorOptions& options) {
  fs::path base_dir = ResolveBasePath(options.endpoint);
  const fs::path listener_path = base_dir / "listener";

  EnsureDirectorySecure(base_dir, kDirectoryMode);
  EnsureFifoSecure(listener_path, kFifoMode);

  const std::string session_id = SanitizeSessionId(
      options.endpoint.empty() ? "demo" : options.endpoint, "demo-session");
  if (MacDebugEnabled()) {
    MacDebugLog("connect endpoint=" + options.endpoint + " base_dir=" + base_dir.string() +
                " session_id=" + session_id);
  }

  const bool infinite_timeout = options.connect_timeout.count() <= 0;
  const auto deadline = infinite_timeout
                            ? std::chrono::steady_clock::time_point::max()
                            : std::chrono::steady_clock::now() + options.connect_timeout;
  auto retry_interval = options.retry_interval.count() <= 0
                            ? std::chrono::milliseconds(50)
                            : options.retry_interval;

  std::map<std::string, Json> handshake_object{
      {"session", Json(session_id)},
      {"pid", Json(static_cast<double>(getpid()))}};
  Json handshake_json(handshake_object);
  std::string handshake_message = handshake_json.dump();
  handshake_message.push_back('\n');

  while (true) {
    int listener_fd = ::open(listener_path.c_str(), O_WRONLY | O_NONBLOCK | O_CLOEXEC);
    if (listener_fd >= 0) {
      WriteAll(listener_fd, handshake_message);
      ::close(listener_fd);
      if (MacDebugEnabled()) {
        MacDebugLog("connect wrote handshake session=" + session_id);
      }
      break;
    }
    if (errno == EINTR) {
      continue;
    }
    if (errno == ENXIO || errno == ENOENT) {
      if (!infinite_timeout && std::chrono::steady_clock::now() >= deadline) {
        throw NamedPipeError("connect named pipe timeout");
      }
      std::this_thread::sleep_for(retry_interval);
      continue;
    }
    ThrowPosixError("open(" + listener_path.string() + ")");
  }

  const fs::path in_path = base_dir / ("session-" + session_id + "-in");
  const fs::path out_path = base_dir / ("session-" + session_id + "-out");

  EnsureFifoSecure(in_path, kFifoMode);
  EnsureFifoSecure(out_path, kFifoMode);

  int write_fd = -1;
  while (true) {
    write_fd = ::open(in_path.c_str(), O_WRONLY | O_CLOEXEC);
    if (write_fd >= 0) {
      break;
    }
    if (errno == EINTR) {
      continue;
    }
    if (errno == ENOENT || errno == ENXIO) {
      if (!infinite_timeout && std::chrono::steady_clock::now() >= deadline) {
        throw NamedPipeError("timeout opening " + in_path.string());
      }
      std::this_thread::sleep_for(retry_interval);
      continue;
    }
    ThrowPosixError("open(" + in_path.string() + ")");
  }

  int read_fd = -1;
  while (true) {
    read_fd = ::open(out_path.c_str(), O_RDONLY | O_CLOEXEC);
    if (read_fd >= 0) {
      break;
    }
    if (errno == EINTR) {
      continue;
    }
    if (errno == ENOENT || errno == ENXIO) {
      if (!infinite_timeout && std::chrono::steady_clock::now() >= deadline) {
        throw NamedPipeError("timeout opening " + out_path.string());
      }
      std::this_thread::sleep_for(retry_interval);
      continue;
    }
    ThrowPosixError("open(" + out_path.string() + ")");
  }

  if (MacDebugEnabled()) {
    MacDebugLog("connect opened session=" + session_id +
                " read_fd=" + std::to_string(read_fd) +
                " write_fd=" + std::to_string(write_fd));
  }

  return std::make_unique<MacPipeHandle>(read_fd, write_fd, out_path, in_path);
}

void ServeNamedPipe(PipeServer& server, PipeListener& listener,
                    NamedPipeHandshake handshake,
                    std::chrono::milliseconds maintenance_interval,
                    NamedPipeShouldContinue should_continue,
                    std::size_t /*worker_threads*/, std::size_t /*read_buffer_size*/) {
  if (!should_continue) {
    throw NamedPipeError("should_continue callback must be provided");
  }
  if (!handshake) {
    throw NamedPipeError("handshake callback must be provided");
  }

  auto* mac_listener = dynamic_cast<MacPipeListener*>(&listener);
  if (!mac_listener) {
    throw NamedPipeError("ServeNamedPipe requires MacPipeListener");
  }

  if (maintenance_interval.count() <= 0) {
    maintenance_interval = std::chrono::milliseconds(100);
  }

  int kq = kqueue();
  if (kq < 0) {
    ThrowPosixError("kqueue");
  }

  int event_pipe[2];
  if (::pipe(event_pipe) != 0) {
    const int err = errno;
    ::close(kq);
    throw PosixError("pipe", err);
  }
  ::fcntl(event_pipe[0], F_SETFL, O_NONBLOCK);
  ::fcntl(event_pipe[1], F_SETFL, O_NONBLOCK);

  auto register_event = [&](uintptr_t ident, int16_t filter, uint16_t flags, uint32_t fflags,
                            intptr_t data, void* udata) {
    struct kevent kev;
    EV_SET(&kev, ident, filter, flags, fflags, data, udata);
    if (kevent(kq, &kev, 1, nullptr, 0, nullptr) != 0) {
      ThrowPosixError("kevent(register)");
    }
  };

  register_event(static_cast<uintptr_t>(event_pipe[0]), EVFILT_READ,
                 EV_ADD | EV_ENABLE, 0, 0, nullptr);

  struct kevent timer_event;
  auto interval_ms = maintenance_interval.count() <= 0 ? 1 : maintenance_interval.count();
  EV_SET(&timer_event, 1, EVFILT_TIMER, EV_ADD | EV_ENABLE, 0,
         static_cast<intptr_t>(interval_ms), nullptr);
  if (kevent(kq, &timer_event, 1, nullptr, 0, nullptr) != 0) {
    const int err = errno;
    ::close(event_pipe[0]);
    ::close(event_pipe[1]);
    ::close(kq);
    throw PosixError("kevent(timer)", err);
  }

  std::mutex pending_mutex;
  std::queue<PipeHandlePtr> pending;
  std::atomic<bool> stop_accept{false};

  TelemetryCenter& telemetry = server.telemetry();
  telemetry.SetGauge("ipc_mac_connections_active", 0.0);
  telemetry.SetGauge("ipc_mac_fifo_sessions_total", 0.0);
  telemetry.SetGauge("ipc_mac_fifo_sessions_active", 0.0);
  telemetry.SetGauge("ipc_mac_fifo_sessions_incomplete", 0.0);
  telemetry.SetGauge("ipc_mac_fifo_sessions_stale", 0.0);

  auto notify_event = [&]() {
    const char byte = 0;
    ::write(event_pipe[1], &byte, 1);
  };

  std::thread accept_thread([&]() {
    while (!stop_accept.load(std::memory_order_acquire)) {
      try {
        auto handle = listener.Accept();
        telemetry.IncrementCounter("ipc_mac_accept_total");
        {
          std::lock_guard<std::mutex> lock(pending_mutex);
          pending.push(std::move(handle));
        }
        notify_event();
      } catch (const std::exception& ex) {
        if (stop_accept.load(std::memory_order_acquire)) {
          break;
        }
        telemetry.IncrementCounter("ipc_mac_accept_error_total");
        server.ReportError(ex.what());
      }
    }
  });

  struct ActiveConnection {
    std::unique_ptr<ServerConnection> connection;
  };

  std::unordered_map<int, ActiveConnection> connections;
  std::atomic<int> active_connections{0};
  std::vector<struct kevent> events(64);

  auto register_connection = [&](int fd, ActiveConnection conn) {
    struct kevent kev;
    EV_SET(&kev, static_cast<uintptr_t>(fd), EVFILT_READ,
           EV_ADD | EV_ENABLE | EV_CLEAR, 0, 0, nullptr);
    if (kevent(kq, &kev, 1, nullptr, 0, nullptr) != 0) {
      server.ReportError("kevent registration failed: " +
                         std::string(std::strerror(errno)));
      return false;
    }
    connections.emplace(fd, std::move(conn));
    const int count = active_connections.fetch_add(1, std::memory_order_relaxed) + 1;
    telemetry.IncrementCounter("ipc_mac_connection_open_total");
    telemetry.SetGauge("ipc_mac_connections_active", static_cast<double>(count));
    return true;
  };

  auto remove_connection = [&](int fd) {
    struct kevent kev;
    EV_SET(&kev, static_cast<uintptr_t>(fd), EVFILT_READ, EV_DELETE, 0, 0, nullptr);
    kevent(kq, &kev, 1, nullptr, 0, nullptr);
    auto it = connections.find(fd);
    if (it != connections.end()) {
      it->second.connection.reset();
      connections.erase(it);
    }
    const int remaining = active_connections.fetch_sub(1, std::memory_order_relaxed) - 1;
    telemetry.IncrementCounter("ipc_mac_disconnect_total");
    telemetry.SetGauge("ipc_mac_connections_active",
                       static_cast<double>(std::max(remaining, 0)));
  };

  auto process_pending = [&]() {
    std::queue<PipeHandlePtr> local;
    {
      std::lock_guard<std::mutex> lock(pending_mutex);
      std::swap(local, pending);
    }

    while (!local.empty()) {
      auto handle = std::move(local.front());
      local.pop();

      SessionContext session;
      bool accepted = false;
      std::string error;
      try {
        accepted = handshake(server, *handle, session);
      } catch (const std::exception& ex) {
        error = ex.what();
      } catch (...) {
        error = "unknown handshake exception";
      }

      if (!accepted) {
        telemetry.IncrementCounter("ipc_mac_handshake_failure_total");
        if (!error.empty()) {
          server.ReportError("handshake failed: " + error);
        }
        continue;
      }

      std::unique_ptr<Channel> channel;
      try {
        channel = server.CreateChannel(std::move(session));
      } catch (const std::exception& ex) {
        telemetry.IncrementCounter("ipc_mac_handshake_failure_total");
        server.ReportError(ex.what());
        continue;
      }

      telemetry.IncrementCounter("ipc_mac_handshake_success_total");

      auto connection = std::make_unique<ServerConnection>(server, std::move(handle),
                                                           std::move(channel));
      auto* pipe_handle = dynamic_cast<MacPipeHandle*>(&connection->handle());
      if (!pipe_handle) {
        server.ReportError("unsupported pipe handle type for macOS ServeNamedPipe");
        continue;
      }
      const int conn_fd = pipe_handle->read_fd();
      if (MacDebugEnabled()) {
        MacDebugLog("serve connection opened fd=" + std::to_string(conn_fd));
      }

      if (!register_connection(conn_fd, ActiveConnection{std::move(connection)})) {
        continue;
      }
    }
  };

  auto drain_health = [&](const fs::path& base_dir) {
    auto stats = linux_diagnostics::InspectFifoHealth(base_dir, kHealthScanInterval);
    telemetry.SetGauge("ipc_mac_fifo_sessions_total", static_cast<double>(stats.total_sessions));
    telemetry.SetGauge("ipc_mac_fifo_sessions_active",
                       static_cast<double>(stats.active_sessions));
    telemetry.SetGauge("ipc_mac_fifo_sessions_incomplete",
                       static_cast<double>(stats.incomplete_sessions));
    telemetry.SetGauge("ipc_mac_fifo_sessions_stale",
                       static_cast<double>(stats.stale_sessions));
    return stats;
  };

  auto base_dir = mac_listener->base_dir();
  linux_diagnostics::FifoHealthStats last_stats;
  auto last_health_scan = std::chrono::steady_clock::now() - kHealthScanInterval;
  auto last_health_alert = std::chrono::steady_clock::time_point::min();

  while (should_continue() && !stop_accept.load(std::memory_order_acquire)) {
    int ready = kevent(kq, nullptr, 0, events.data(), static_cast<int>(events.size()), nullptr);
    if (ready < 0) {
      if (errno == EINTR) {
        continue;
      }
      const int err = errno;
      stop_accept.store(true, std::memory_order_release);
      mac_listener->SignalInterrupt();
      notify_event();
      if (accept_thread.joinable()) {
        accept_thread.join();
      }
      ::close(event_pipe[0]);
      ::close(event_pipe[1]);
      ::close(kq);
      throw PosixError("kevent", err);
    }

    for (int i = 0; i < ready; ++i) {
      const auto& ev = events[i];
      if (ev.filter == EVFILT_READ && static_cast<int>(ev.ident) == event_pipe[0]) {
        DrainPipe(event_pipe[0]);
        process_pending();
        continue;
      }

      if (ev.filter == EVFILT_TIMER) {
        for (auto& entry : connections) {
          entry.second.connection->FlushMaintenance();
        }

        const auto now = std::chrono::steady_clock::now();
        if (now - last_health_scan >= kHealthScanInterval) {
          last_health_scan = now;

          const std::size_t repaired =
              linux_diagnostics::CleanupFifoSessions(base_dir, kHealthAlertInterval);
          if (repaired > 0) {
            telemetry.IncrementCounter("ipc_mac_fifo_repairs_total",
                                       static_cast<uint64_t>(repaired));
            std::ostringstream msg;
            msg << "mac fifo cleanup removed " << repaired << " stale session(s)";
            server.ReportError(msg.str());
          }

          auto stats = drain_health(base_dir);
          const bool anomalous = stats.incomplete_sessions > 0 || stats.stale_sessions > 0;
          const bool changed = stats.incomplete_sessions != last_stats.incomplete_sessions ||
                                stats.stale_sessions != last_stats.stale_sessions;
          if (anomalous && (changed ||
                             now - last_health_alert >= kHealthAlertInterval)) {
            std::ostringstream oss;
            oss << "mac fifo health: total=" << stats.total_sessions
                << " active=" << stats.active_sessions
                << " incomplete=" << stats.incomplete_sessions
                << " stale=" << stats.stale_sessions;
            server.ReportError(oss.str());
            last_health_alert = now;
          }
          last_stats = stats;
        }
        continue;
      }

      int fd = static_cast<int>(ev.ident);
      auto it = connections.find(fd);
      if (it == connections.end()) {
        continue;
      }

      if ((ev.flags & EV_EOF) != 0 || (ev.flags & EV_ERROR) != 0) {
        if (MacDebugEnabled()) {
          MacDebugLog("serve event eof/error fd=" + std::to_string(fd));
        }
        remove_connection(fd);
        continue;
      }

      bool keep = true;
      try {
        keep = it->second.connection->PumpOnce();
      } catch (const std::exception& ex) {
        server.ReportError(ex.what());
        keep = false;
      }

      if (!keep) {
        if (MacDebugEnabled()) {
          MacDebugLog("serve pump closed fd=" + std::to_string(fd));
        }
        remove_connection(fd);
      }
    }
  }

  stop_accept.store(true, std::memory_order_release);
  mac_listener->SignalInterrupt();
  notify_event();
  if (accept_thread.joinable()) {
    accept_thread.join();
  }

  DrainPipe(event_pipe[0]);

  for (auto& entry : connections) {
    remove_connection(entry.first);
  }
  connections.clear();

  ::close(event_pipe[0]);
  ::close(event_pipe[1]);
  ::close(kq);
}

}  // namespace asycipc

#endif

#if !defined(_WIN32) && !defined(__linux__) && !defined(__APPLE__)
namespace asycipc {

std::unique_ptr<PipeListener> CreateNamedPipeListener(
    const NamedPipeListenerOptions& options) {
  (void)options;
  throw NamedPipeError("macOS pipes not supported on this platform");
}

PipeHandlePtr ConnectNamedPipeEndpoint(
    const NamedPipeConnectorOptions& options) {
  (void)options;
  throw NamedPipeError("macOS pipes not supported on this platform");
}

void ServeNamedPipe(PipeServer&, PipeListener&, NamedPipeHandshake,
                    std::chrono::milliseconds, NamedPipeShouldContinue,
                    std::size_t, std::size_t) {
  throw NamedPipeError("macOS pipes not supported on this platform");
}

}  // namespace asycipc

#endif

